package seven.three;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import seven.entity.OrderBean;

public class GroupSort {
	static class SortMapper extends Mapper<LongWritable,Text,OrderBean,NullWritable>{
		OrderBean bean  = new OrderBean();
		@Override
		protected void map(LongWritable key, Text value,
				Mapper<LongWritable, Text, OrderBean, NullWritable>.Context context)
				throws IOException, InterruptedException {
			 String line = value.toString();
			 String[] feilds = line.split(",");
			 bean.set(new Text(feilds[0]), new DoubleWritable(Double.parseDouble(feilds[2])));
			 context.write(bean,NullWritable.get());
			 
		}  
	}
	static class SortReudcer extends Reducer<OrderBean,NullWritable,OrderBean,NullWritable>{

		@Override
		protected void reduce(OrderBean key, Iterable<NullWritable> arg1,
				Reducer<OrderBean, NullWritable, OrderBean, NullWritable>.Context context)
				throws IOException, InterruptedException {
			 context.write(key,NullWritable.get());
		}
	}
	
	public static void main(String[] args)throws Exception {
		Configuration conf = new Configuration();
		Job job = Job.getInstance(conf);
		job.setJarByClass(GroupSort.class);
		
		//任务输出类型
		job.setOutputKeyClass(OrderBean.class);
		job.setOutputValueClass(NullWritable.class);
		
		//指定map reduce
		job.setMapperClass(SortMapper.class);
		job.setReducerClass(SortReudcer.class);
		
		
		job.setGroupingComparatorClass(MyGroupingComparator.class);
		job.setPartitionerClass(ItemIdPartitioner.class);
		job.setNumReduceTasks(2);
		
		FileInputFormat.setInputPaths(job,new Path("E:/code/workspace/hadoopone/src/main/java/seven/entity/input/"));
		//指定job输出结果目录所在
		FileOutputFormat.setOutputPath(job, new Path("E:/code/workspace/hadoopone/src/main/java/seven/entity/output/"));
		
		job.waitForCompletion(true);
		
		
		
	}
}
